package org.example;

import com.google.gson.Gson;
import com.google.gson.internal.LinkedTreeMap;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.*;
import java.util.HashMap;

public class MysqlSink extends RichSinkFunction<String> {
    Connection connection;
    PreparedStatement iStmt;
    PreparedStatement dStmt;
    PreparedStatement uStmt;
    final String URL = "XXX";
    final String USERNAME = "XXX";
    final String PASSWORD = "XXX";
    final String DB = "XXX";
    final String TABLE = "XXX";


    private Connection getConnection() {
        Connection conn = null;
        try {
            conn = DriverManager.getConnection(URL, USERNAME, PASSWORD);
       } catch (Exception e) {
            e.printStackTrace();
       }
        return conn;
   }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = getConnection();
        String insertSql = "insert into T_DP_N_CommentImage(ImageId,ObjectId,CommentId,CreateTime,TagId,AuditStatus,AuditUserId,AuditTime,TagName,ImagePicScore,IsPick,IsRepeat,Level) values (?,?,?,?,?,?,?,?,?,?,?,?,?)";
        String deleteSql = "delete from T_DP_N_CommentImage where ImageId = ?";
        String updateSql = "update T_DP_N_CommentImage set ObjectId = ?,CommentId = ?,CreateTime = ?,TagId = ?,AuditStatus = ?,AuditUserId = ?,AuditTime = ?,TagName = ?,ImagePicScore = ?,IsPick = ?,IsRepeat = ?,Level = ? where ImageId = ?";
        iStmt = connection.prepareStatement(insertSql);
        dStmt = connection.prepareStatement(deleteSql);
        uStmt = connection.prepareStatement(updateSql);
        System.out.println("open数据库连接....");
   }

    @Override
    public void invoke(String value, Context context) throws Exception {
        Gson t = new Gson();
        HashMap<String, Object> hs = t.fromJson(value, HashMap.class);
        LinkedTreeMap<String, Object> source = (LinkedTreeMap<String, Object>) hs.get("source");
        String db = (String) source.get("db");
        String table = (String) source.get("table");
        String op = (String) hs.get("op");

        if (DB.equals(db) && TABLE.equals(table)) {
            LinkedTreeMap<String, Object> data = (LinkedTreeMap<String, Object>) hs.get("after");
            LinkedTreeMap<String, Object> before = (LinkedTreeMap<String, Object>) hs.get("before");
            // 删除数据
            if (before != null && data == null) {
                String imageId = (String) before.get("ImageId");
                dStmt.setString(1, imageId);
                System.out.println("删除数据....");

                dStmt.executeUpdate();
           } else {
                String imageId = (String) data.get("ImageId");
                String objectId = (String) data.get("ObjectId");
                Double commentId = (Double) data.get("CommentId");
                String createTime = (String)data.get("CreateTime");
                Double tagId = ((Double) data.get("TagId"));
                Double auditStatus = ((Double) data.get("AuditStatus"));
                String auditUserId = (String) data.get("AuditUserId");
                String auditTime = (String) data.get("AuditTime");
                String tagName = (String) data.get("TagName");
                Double imagePicScore = (Double) data.get("ImagePicScore");
                Double isPick = ((Double) data.get("IsPick"));
                Double isRepeat = (Double) data.get("IsRepeat");
                String level = (String) data.get("Level");
                // 执行插入
                if ("r".equals(op) || "c".equals(op)) {
                    iStmt.setString(1, imageId);
                    iStmt.setString(2, objectId);
                    if (commentId != null) {
                        iStmt.setInt(3, commentId.intValue());
                   } else {
                        iStmt.setNull(3, Types.INTEGER);
                   }
                    if (createTime != null) {
                        iStmt.setString(4, createTime);
                   } else {
                        iStmt.setNull(4, Types.TIMESTAMP);
                   }
                    if (tagId != null) {
                        iStmt.setInt(5, tagId.intValue());
                   } else {
                        iStmt.setNull(5, Types.INTEGER);
                   }
                    if (auditStatus != null) {
                        iStmt.setInt(6, auditStatus.intValue());
                   } else {
                        iStmt.setNull(6, Types.INTEGER);
                   }
                    iStmt.setString(7, auditUserId);
                    if (auditTime != null) {
                        iStmt.setString(8, auditTime);
                   } else {
                        iStmt.setNull(8, Types.TIMESTAMP);
                   }
                    iStmt.setString(9, tagName);
                    if (imagePicScore != null) {
                        iStmt.setInt(10, imagePicScore.intValue());
                   } else {
                        iStmt.setNull(10, Types.INTEGER);
                   }
                    if (isPick != null) {
                        iStmt.setInt(11, isPick.intValue());
                   } else {
                        iStmt.setNull(11, Types.TINYINT);
                   }
                    if (isRepeat != null) {
                        iStmt.setInt(12, isRepeat.intValue());
                   } else {
                        iStmt.setNull(12, Types.TINYINT);
                   }
                    iStmt.setString(13, level);
                    System.out.println("插入数据....");

                    iStmt.executeUpdate();
               } /* 更新 */ else if ("u".equals(op)) {
                    uStmt.setString(1, objectId);
                    if (commentId != null) {
                        uStmt.setInt(2, commentId.intValue());
                   } else {
                        uStmt.setNull(2, Types.INTEGER);
                   }
                    if (createTime != null) {
                        uStmt.setString(3, createTime);
                   } else {
                        uStmt.setNull(3, Types.TIMESTAMP);
                   }
                    if (tagId != null) {
                        uStmt.setInt(4, tagId.intValue());
                   } else {
                        uStmt.setNull(4, Types.INTEGER);
                   }
                    if (auditStatus != null) {
                        uStmt.setInt(5, auditStatus.intValue());
                   } else {
                        uStmt.setNull(5, Types.INTEGER);
                   }
                    uStmt.setString(6, auditUserId);
                    if (auditTime != null) {
                        uStmt.setString(7, auditTime);
                   } else {
                        uStmt.setNull(7, Types.TIMESTAMP);
                   }
                    uStmt.setString(8, tagName);
                    if (imagePicScore != null) {
                        uStmt.setInt(9, imagePicScore.intValue());
                   } else {
                        uStmt.setNull(9, Types.INTEGER);
                   }
                    if (isPick != null) {
                        uStmt.setInt(10, isPick.intValue());
                   } else {
                        uStmt.setNull(10, Types.TINYINT);
                   }
                    if (isRepeat != null) {
                        uStmt.setInt(11, isRepeat.intValue());
                   } else {
                        uStmt.setNull(11, Types.TINYINT);
                   }
                    uStmt.setString(12, level);
                    uStmt.setString(13, imageId);
                    System.out.println("修改数据....");

                    uStmt.executeUpdate();
               }
           }
       }
   }

    @Override
    public void close() throws Exception {
        System.out.println("close数据库连接....");
        if (iStmt != null) {
            iStmt.close();
       }
        if (dStmt != null) {
            dStmt.close();
       }
        if (uStmt != null) {
            uStmt.close();
       }
        if (connection != null) {
            connection.close();
       }
   }
}